"""
Case Type   : 连接
Case Name   : ssl参数使用gs_guc set设置，并校验预期结果
Create At   : 2025/4/3
Owner       : @li-xin12345
Description :
    1、查看ssl相关参数默认值；
    show ssl;
    show ssl_cert_file;
    show ssl_key_file;
    show ssl_ca_file;
    2、创建用户
    drop user if exists user166 cascade;
    create user user166 password '{macro.COMMON_PASSWD}';
    grant all privileges to user166;
    3、使用create_ca.sh脚本生成证书
    4、修改pg_hba.conf文件
    hostssl    all    user166    {self.user_node.db_host}/32    cert
    5、修改postgresql.conf文件
    gs_guc set -N all -I all -c "ssl=on"
    gs_guc set -N all -I all -c "ssl_cert_file='server.crt'"
    gs_guc set -N all -I all -c "ssl_key_file='server.key'"
    gs_guc set -N all -I all -c "ssl_ca_file='cacert.pem'"
    6、配置env文件
    export PGSSLCERT="~/test/client.crt"
    export PGSSLKEY="~/test/client.key"
    export PGSSLMODE="verify-ca"
    export PGSSLROOTCERT="~/test/cacert.pem"
    7、使用ssl连接免密成功
    source {macro.DB_ENV_PATH};
    gsql -d postgres -p {self.user_node.db_port}
    -h {self.user_node.db_host} -U user166 -c "show ssl;"
    8、修改相关参数
    gs_guc set -N all -I all -c "ssl=off"
    9、使用ssl连接免密失败
    source {macro.DB_ENV_PATH};
    gsql -d postgres -p {self.user_node.db_port}
    -h {self.user_node.db_host} -U user166 -c "select 1;"
    10、恢复环境 恢复env文件 恢复postgresql.conf文件 恢复pg_hba.conf文件
Expect      :
    1、显示默认值；
    2、创建用户成功
    3、证书生成成功
    4、添加pg_hba.conf成功
    5、配置postgresql.conf成功
    6、配置env文件成功
    7、使用ssl连接免密成功
    8、修改参数并重启成功
    9、使用ssl连接免密失败
    10、恢复环境成功
History     :
"""
import os
import subprocess
import unittest

from yat.test import Node
from yat.test import macro

from testcase.CONNECTORS.PDBC.PSYCOPG2. \
    Opengauss_Function_Connect_Psycopg_Common import Psycopg2Common
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger

pri_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(3 != pri_sh.get_node_num(), '非1+2环境不执行')
class SessionTest(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        text = f'-----{os.path.basename(__file__)} start-----'
        self.log.info(text)
        self.constant = Constant()
        self.common = Common()
        self.conn_info = {'dbname': pri_sh.node.db_name,
                          'user': pri_sh.node.db_user,
                          'password': pri_sh.node.db_password,
                          'host': pri_sh.node.ssh_host,
                          'port': pri_sh.node.db_port}
        self.psycopg_com = Psycopg2Common(**self.conn_info)
        self.pri_user_node = Node('PrimaryDbUser')
        self.script_name = 'create_ca.sh'
        self.remote_path = os.path.join('/home', pri_sh.node.ssh_user)
        self.zip_name = 'db-cert-replacement.zip'
        self.ssl_params = ['ssl', 'ssl_cert_file', 'ssl_key_file',
                           'ssl_ca_file']
        self.user_name = 'u_connect_0166'

        text = '----前置操作: 备份白名单 expect: 备份成功----'
        self.log.info(text)
        res = self.psycopg_com.bak_conf(bak_flag=True)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

        text = '----前置操作: 备份配置文件 expect: 备份成功----'
        self.log.info(text)
        res = self.psycopg_com.bak_conf(bak_flag=True,
                                        f_name=macro.DB_PG_CONFIG_NAME)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

        text = '----前置操作: 备份环境变量 expect: 备份成功----'
        self.log.info(text)
        res = self.psycopg_com.bak_conf(bak_flag=True,
                                        f_name=macro.DB_ENV_PATH)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

    def test_ssl(self):
        text = '----step1: 查看ssl相关参数默认值；expect: 显示默认值----'
        self.log.info(text)
        for i in self.ssl_params:
            res = self.common.show_param(i)
            self.assertTrue(res)

        text = '----step2: 创建用户；expect: 创建用户成功----'
        self.log.info(text)
        cmd = f"drop user if exists {self.user_name} cascade;" \
              f"create user {self.user_name} " \
              f"password '{macro.COMMON_PASSWD}';" \
              f"grant all privileges to {self.user_name};"
        res = pri_sh.execut_db_sql(cmd)
        self.log.info(res)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.CREATE_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')
        self.assertIn(self.constant.ALTER_ROLE_SUCCESS_MSG, res,
                      f'执行失败: {text}')

        text = '----step3: 使用create_ca.sh脚本生成证书；expect: 证书生成成功----'
        self.log.info(text)
        text = '---将create_ca脚本发送到数据库主机---'
        self.log.info(text)
        self.common.scp_file(pri_sh.node, self.script_name,
                             self.remote_path)
        res = pri_sh.node.sh(
            f'ls {self.remote_path}|grep {self.script_name}').result()
        self.log.info(res)
        self.assertTrue(res.strip(), f'执行失败: {text}')

        text = '---通过create_ca脚本生成证书---'
        self.log.info(text)
        sh_cmd = f'source {macro.DB_ENV_PATH};' \
                 f'sh {self.script_name} {pri_sh.node.db_user} ' \
                 f'{pri_sh.node.ssh_password} {self.user_name}'
        self.log.info(sh_cmd)
        res = pri_sh.node.sh(sh_cmd).result()
        self.log.info(res)
        self.log.info(self.remote_path)
        res = pri_sh.node.sh(
            f'ls {self.remote_path}|grep {self.zip_name}').result()
        self.log.info(res)
        self.assertTrue(res.strip(), f'执行失败: {text}')

        text = '---替换证书---'
        self.log.info(text)
        zip_path = os.path.join(self.remote_path, self.zip_name)
        self.log.info(zip_path)
        replace_cmd = f'source {macro.DB_ENV_PATH};' \
                      f'gs_om -t cert --cert-file={zip_path}'
        self.log.info(replace_cmd)
        replace_res = pri_sh.node.sh(replace_cmd).result()
        self.log.info(replace_res)
        self.assertIn('Successfully replace om ssl cert files', replace_res,
                      f'执行失败: {text}')
        self.assertIn('Successfully distributed cert files on all nodes',
                      replace_res, f'执行失败: {text}')

        is_started = pri_sh.restart_db_cluster()
        self.assertTrue(is_started, f'执行失败: {text}')

        text = '----step4: 修改pg_hba.conf文件；expect: 成功----'
        self.log.info(text)
        res = self.psycopg_com.set_pghba(open_flag=True,
                                         mode='cert')
        self.assertTrue(res, f'执行失败: {text}')

        param = f'hostssl all all {pri_sh.node.ssh_host}/32 cert'
        res = pri_sh.execute_gsguc('reload', self.constant.GSGUC_SUCCESS_MSG,
                                   '', pghba_param=param)
        self.log.info(res)
        self.assertTrue(res, f'执行失败: {text}')

        text = '----step5: 修改postgresql.conf文件;' \
               'expect:配置postgresql.conf成功----'
        self.log.info(text)
        for param in (f"ssl_cert_file='{self.remote_path}/client.crt'",
                      f"ssl_key_file='{self.remote_path}/client.key'",
                      f"ssl_ca_file='{self.remote_path}/cacert.pem'",
                      "ssl=on"):
            res = pri_sh.execute_gsguc('set',
                                            self.constant.GSGUC_SUCCESS_MSG,
                                            param,
                                            single=True)
            self.assertTrue(res, f'执行失败: {text}')
        is_started = pri_sh.restart_db_cluster()
        self.assertTrue(is_started, f'执行失败: {text}')

        text = '----step6: 配置env文件;expect: 配置env文件成功----'
        self.log.info(text)
        export_content = f'''\
        export PGSSLCERT="{self.remote_path}/client.crt"
        export PGSSLKEY="{self.remote_path}/client.key"
        export PGSSLMODE="verify-ca"
        export PGSSLROOTCERT="{self.remote_path}/cacert.pem"
        '''

        cmd = f'cat << \'EOF\' >> {macro.DB_ENV_PATH}\n{export_content}\nEOF'

        res = pri_sh.node.sh(cmd).result()
        self.log.info(f"命令输出:\n{res}")

        verify_cmd = f'tail -n5 {macro.DB_ENV_PATH}'
        verify_res = pri_sh.node.sh(verify_cmd).result()
        self.log.info(f"文件验证结果:\n{verify_res}")

        text = '----step7: 使用ssl连接免密成功;expect: 使用ssl连接免密成功----'
        self.log.info(text)
        cmd = (f'source {macro.DB_ENV_PATH};'
               f'gsql -d postgres -p {pri_sh.node.db_port} '
               f'-h {pri_sh.node.db_host} -U {self.user_name} -c "show ssl;"')
        self.log.info(cmd)
        res = pri_sh.node.sh(cmd).result()
        self.log.info(res)
        self.assertNotIn('ERROR', res, f'执行失败: {text}')
        self.assertIn('on', res, f'执行失败: {text}')

        text = '----step8: 修改相关参数;expect: 修改参数并重启成功----'
        self.log.info(text)
        res = pri_sh.execute_gsguc(
            'set', self.constant.GSGUC_SUCCESS_MSG, 'ssl=off')
        self.log.info(res)
        self.assertTrue(res)
        res = pri_sh.restart_db_cluster()
        self.log.info(res)

        text = '----step9: 使用ssl连接免密失败;expect: 使用ssl连接免密失败----'
        self.log.info(text)
        cmd = f'''source {macro.DB_ENV_PATH};
        gsql -d postgres -p {pri_sh.node.db_port} -h {pri_sh.node.db_host} 
        -U {self.user_name} -c "show ssl;"
        '''
        self.log.info(cmd)
        res = pri_sh.node.sh(cmd).result()
        self.log.info(res)

    def tearDown(self):
        text = '----step10: 恢复环境 恢复env文件 ' \
               '恢复postgresql.conf文件 恢复pg_hba.conf文件;expect:恢复环境成功----'
        self.log.info(text)
        text1 = f'----后置操作：删除解压文件目录----'
        self.log.info(text1)
        cmd = f'if [ -e {self.remote_path}/client* ] || ' \
              f'[ -e {self.remote_path}/server.* ] || ' \
              f'[ -e {self.remote_path}/cacert.pem ]; ' \
              f'then ' \
              f'rm -rf {self.remote_path}/client* {self.remote_path}/server.* ' \
              f'{self.remote_path}/cacert.pem; ' \
              f'fi'
        res1 = self.pri_user_node.sh(cmd)
        self.log.info(res1)

        text2 = f'----后置操作：还原白名单----'
        self.log.info(text2)
        res2 = self.psycopg_com.bak_conf(bak_flag=False)

        text3 = f'----后置操作：还原配置文件----'
        self.log.info(text3)
        res3 = self.psycopg_com.bak_conf(bak_flag=False,
                                         f_name=macro.DB_PG_CONFIG_NAME)

        text4 = f'----后置操作：删除生成证书----'
        self.log.info(text4)
        rm_cmd = f'rm -rf {self.script_name} {self.zip_name} cacert.pem ' \
                 f'client.* server.* demoCA openssl.cnf'
        res4 = pri_sh.node.sh(rm_cmd).result()
        self.log.info(res4)
        rm_cmd = rm_cmd.replace(self.script_name, '')
        res5 = subprocess.getoutput(rm_cmd)
        self.log.info(res5)

        text6 = f'----后置操作：删除用户----'
        self.log.info(text6)
        cmd = f'drop user if exists {self.user_name} cascade;'
        res6 = pri_sh.execut_db_sql(cmd)
        self.log.info(res6)

        text7 = f'----后置操作：恢复环境变量----'
        self.log.info(text7)
        res7 = self.psycopg_com.bak_conf(bak_flag=False,
                                        f_name=macro.DB_ENV_PATH)
        self.log.info(res7)

        self.assertTrue(res1, f'执行失败: {text1}')
        self.assertTrue(res2, f'执行失败: {text2}')
        self.assertTrue(res3, f'执行失败: {text3}')
        self.assertEqual(len(res4.strip()), 0, f'执行失败: {text4}')
        self.assertEqual(len(res5.strip()), 0, f'执行失败: {text4}')
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG, res6,
                      f'执行失败: {text6}')
        self.assertTrue(res7, f'执行失败: {text7}')

        text = f'-----{os.path.basename(__file__)} end-----'
        self.log.info(text)
